package com.sky.mqtt;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

/**
 * @Content: MQTT消费端
 * @Author: LS
 * @Date: 2023/10/13 15:25
 */
@Configuration
@IntegrationComponentScan
public class MQTTInboundConfiguration {
    @Autowired
    private MQTTProperties mqttProperties;
    @Autowired
    private MQTTOutboundConfiguration mqttOutboundConfiguration;
    /**
     * 接收端
     */

    //配置client,监听topic
    @Bean
    public MessageProducer inbound(){
        String[] inboundTopics = mqttProperties.getDefaultTopic().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getClientId()+"_inbound",
                mqttOutboundConfiguration.mqttPahoClientFactory(),
                inboundTopics
        );//对inboundTopics主题进行监听
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setOutputChannel(mqttOutboundConfiguration.mqttOutboundChannel());
        return adapter;
    }

    //通过通道获取数据
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler handler(){
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println("-------------------");
                System.out.println("message:"+message.getPayload());
                System.out.println("PacketID:"+message.getHeaders().getId());
                System.out.println("Qos:"+message.getHeaders().get(MqttHeaders.QOS));
                String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
                System.out.println("topic:"+topic);
            }
        };
    }
}
